package com.gitee.dufafei.spark.listener

import com.gitee.dufafei.spark.pattern.Logging
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.util.QueryExecutionListener

/**
 * spark对sql的操作会形成一个data frame，data frame中的logical包含了sql的语法树
 * 通过对logical的语法树解析可以获取当前stage所操作的输入表和输出表
 * 将整套表关系链连接起来再去除中间表(createOrReplaceTempView)即可获取当前作业的输入表和输出表信息
 */
class ISparkSqlListener extends QueryExecutionListener with Logging {

  override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
    val sparkSqlParser = new ISparkSqlParser(qe)
    val tables = sparkSqlParser.resolveLogicPlan()

    val inputTables = tables._1
    val outputTables = tables._2
    inputTables.foreach{ source =>
      outputTables.foreach{ target =>
        val tableEdge = TableEdge(source, target)
        println(tableEdge)
      }
    }
  }

  override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = {}
}
